package com.my.hbase.util;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.List;
import com.my.hbase.demo.User;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 *  HBase映射工具类
 *  @author  liujk
 *  @version 2016-07-14
 */
public class HBaseUtil {

    private static Logger logger = LoggerFactory.getLogger(HBaseUtil.class);


    /**
     * JavaBean转换为Put
     *
     * @param obj
     * @return
     * @throws Exception
     */
    public static Put bean2Put(Object obj) throws Exception {
        Put put = new Put(Bytes.toBytes(parseObjRowKey(obj)));
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();

        for (Field field : fields) {
            if (!field.isAnnotationPresent(HBaseColumn.class)) {
                continue;
            }

            field.setAccessible(true);
            HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
            String family = orm.family();
            String qualifier = orm.qualifier();
            boolean rowkey = orm.rowkey();

            if(rowkey){
                continue;
            }

            if (StringUtils.isBlank(family) || StringUtils.isBlank(qualifier)) {
                continue;
            }

            Object fieldObj = field.get(obj);
            //TODO package value
            if (fieldObj.getClass().isArray()) {
                System.out.println("show this....");
            }

            if (field.get(obj) != null || StringUtils.isNotBlank(field.get(obj).toString())) {

                String type = field.getType().getName();

                if("java.lang.String".equals(type)){
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(field.get(obj).toString()));
                }else if("int".equals(type) || "java.lang.Integer".equals(type)){
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes((Integer)field.get(obj)));
                }else if("long".equals(type) || "java.lang.Long".equals(type)){
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes((Long)field.get(obj)));
                }else if("[B".equals(type)){ //字节
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), (byte[])field.get(obj));
                }else if("double".equals(type) || "java.lang.Double".equals(type)){
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes((Double)field.get(obj)));
                }else if("float".equals(type) || "java.lang.Float".equals(type)){
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes((Float)field.get(obj)));
                } else{
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), (byte[])field.get(obj));
                }

            }
        }
        return put;
    }



    /**
     * HBase result 转换为 bean
     * @param result
     * @param obj
     * @return
     * @throws Exception
     */
    public static Object result2Bean(Result result, Object obj) throws Exception {
        if (result == null) {
            return null;
        }
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            if (!field.isAnnotationPresent(HBaseColumn.class)) {
                continue;
            }
            HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
            String family = orm.family();
            String qualifier = orm.qualifier();
            boolean rowkey = orm.rowkey();

            if(!rowkey){
                if (StringUtils.isBlank(family) || StringUtils.isBlank(qualifier)) {
                    continue;
                }
            }

            String fieldName = field.getName();
            Object value;
            if (rowkey) {
                value = new String(result.getRow());
            } else {
                String type = field.getType().getName();

                value = getResultValueByType(result, family, qualifier, type);
            }
            String firstLetter = fieldName.substring(0, 1).toUpperCase();
            String setMethodName = "set" + firstLetter + fieldName.substring(1);
            Method setMethod = clazz.getMethod(setMethodName, new Class[] { field.getType() });
            setMethod.invoke(obj, new Object[] { value });
        }
        return obj;
    }

    public static String getTableName(Object obj){
        Class<?> clazz = obj.getClass();
        HBaseTable hBaseTable = clazz.getAnnotation(HBaseTable.class);

        String tableName = hBaseTable.tableName();

        return tableName;
    }

    public static String[] getFamilies(Object obj){
        Class<?> clazz = obj.getClass();
        HBaseTable hBaseTable = clazz.getAnnotation(HBaseTable.class);

        String families = hBaseTable.families();

        if(families.contains(",")){
            return families.split(",");
        }else{
            return new String[]{families};
        }
    }

    /**
     * TODO add other type value
     * @param result
     * @param family
     * @param qualifier
     * @return
     */
    private static Object getResultValueByType(Result result, String family, String qualifier, String type) {

        List<Cell> cells = result.getColumnCells(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        if (cells.size() == 1) {
            Cell cell = cells.get(0);

            if("java.lang.String".equals(type)){
                return  Bytes.toString(CellUtil.cloneValue(cell));
            }else if("int".equals(type) || "java.lang.Integer".equals(type)){
                return Bytes.toInt( CellUtil.cloneValue(cell));
            }else if("long".equals(type) || "java.lang.Long".equals(type)){
                return  Bytes.toLong(CellUtil.cloneValue(cell));
            }else if("[B".equals(type)){ //字节
                return  CellUtil.cloneValue(cell);
            }else if("double".equals(type) || "java.lang.Double".equals(type)){
                return Bytes.toDouble(CellUtil.cloneValue(cell));
            } else if("float".equals(type) || "java.lang.Float".equals(type)){
                return Bytes.toFloat(CellUtil.cloneValue(cell));
            }else{
                return  CellUtil.cloneValue(cell);
            }
        }
        return "";
    }

    /**
     * 获取rowkey
     * @param obj
     * @return
     */
    public static String parseObjRowKey(Object obj) {
        Class<?> clazz = obj.getClass();

        try {
            Field[] fields = clazz.getDeclaredFields();

            for (Field field : fields) {
                if (!field.isAnnotationPresent(HBaseColumn.class)) {
                    continue;
                }
                HBaseColumn orm = field.getAnnotation(HBaseColumn.class);
                boolean rowkey = orm.rowkey();

                if(rowkey){
                    field.setAccessible(true);
                    Object object = field.get(obj);
                    return object.toString();
                }
            }

        }  catch (Exception e) {
            logger.warn("", e);
        }
        return "";
    }

    public static  void main(String arg0[]){

        /*
           获取table名称
           User user = new User();
           System.out.println(HBaseUtil.getTableName(user));
        */

        User user = new User();

        System.out.println("##############获取列族开始###########");
       String[] families = HBaseUtil.getFamilies(user);
        for(String family : families){
            System.out.println("列族是:" + family);
        }

        System.out.println("#############获取rowkey开始################");

        user.setRowkey("234535@beijing@changjiang");

        System.out.println("rowkey是："+HBaseUtil.parseObjRowKey(user));

    }



}
